package com.tdy.cdc.app.dwd;

import com.tdy.cdc.app.BaseSqlApp;
import com.tdy.cdc.common.Constant;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.FlinkSourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamStatementSet;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: NanHuang
 * @Date: 2023/04/03/10:38
 * @Description:
 */
public class App_01_04 extends BaseSqlApp {
    public static void main(String[] args) {
        new App_01_04().init(
                4001,
                "App_01_04",
                2
        );
    }
    @Override
    protected void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 创建源表
        createSourceTables(tableEnv);
        // 2 join
        joinTables(tableEnv);
        // 3 写入Kafka
        writeToKafka(tableEnv);
    }

    private void writeToKafka(StreamTableEnvironment tableEnv) {
        // 创建Kafka映射表
        tableEnv.executeSql("create table target1 (" +
                "RECORD_ID                      STRING," +
                "HOSPITAL_NAME                  STRING," +
                "HOSPITAL_CODE                  STRING," +
                "AGE_UNIT                       STRING," +
                "AGE                            STRING," +
                "SEX_NAME                       STRING," +
                "SEX_CODE                       STRING," +
                "ID_CARD_NUMBER                 STRING," +
                "PATIENT_VISIT_CATEGORY_CODE    STRING," +
                "PATIENT_VISIT_CATEGORY         STRING," +
                "OUTPATIENT_VISIT_FLOW_ID       STRING," +
                "OUTPATIENT_NO                  STRING," +
                "PATIENT_ID                     STRING," +
                "PATIENT_NAME                   STRING," +
                "DOMAIN_ID                      STRING," +
                "TELECOM                        STRING," +
                "VISIT_DATE                     STRING," +
                "VISIT_TIMES                    STRING," +
                "APPLY_SHEET_NO                 STRING," +
                "BIRTH_DATE                     STRING," +
                "DEPT_ID                        STRING," +
                "DEPT_NAME                      STRING," +
                "AUTHOR_CODE                    STRING," +
                "AUTHOR_NAME                    STRING," +
                "CUSTODIAN_CODE                 STRING," +
                "CUSTODIAN_NAME                 STRING," +
                "RESPONSE_TIME                  STRING," +
                "RESPONSE_CODE                  STRING," +
                "RESPONSE_NAME                  STRING," +
                "ALLERGIC_HISTORY_FLAG          STRING," +
                "ALLERGIC_HISTORY               STRING," +
                "CHIEF_COMPLAINT                STRING," +
                "PRESENT_ILLNESS_HISTORY        STRING," +
                "PAST_ILLNESS_HISTORY           STRING," +
                "PHYSICAL_EXAM                  STRING," +
                "FIRST_VISIT_FLAG               STRING," +
                "PD_NAME                        STRING," +
                "CHINA_MED_OBSERVE_RESULT       STRING," +
                "DISCRIMINATE_BASIS             STRING," +
                "RULES_CODE                     STRING," +
                "RESCUE_START_DATE              STRING," +
                "RESCUE_END_DATE                STRING," +
                "RESCUERS_LIST                  STRING," +
                "PAT_CATEGORY_CODE              STRING," +
                "PAT_CATEGORY                   STRING," +
                "INCOME_OBSERVATORY_DATE        STRING," +
                "EMERGENCY_COURSE               STRING," +
                "NEEDING_ATTENTION              STRING," +
                "DESTINATION_CODE               STRING," +
                "EMERGENCY_RECORD               STRING," +
                "DIAG_CODE                      STRING," +
                "DIAG_NAME                      STRING," +
                "CHINA_DISEASE_CODE             STRING," +
                "CHINA_DISEASE_NAME             STRING," +
                "CHINA_SYNDROME_CODE            STRING," +
                "CHINA_SYNDROME_NAME            STRING," +
                "ORDERS_AUDIT_STAFF_ID          STRING," +
                "ORDERS_AUDIT_TIME              STRING," +
                "ORDERS_AUDIT_STAFF_NAME        STRING," +
                "ORDERS_CANCEL_DOCTOR_ID        STRING," +
                "ORDERS_CANCEL_TIME             STRING," +
                "ORDERS_CANCEL_DOCTOR_NAME      STRING," +
                "ORDERS_EXEC_STAFF_ID           STRING," +
                "ORDERS_EXEC_STAFF_NAME         STRING," +
                "ORDERS_EXEC_TIME               STRING," +
                "ORDER_EXEC_DEPT                STRING," +
                "EXECUTOR_STATUS                STRING," +
                "ORDERS_CREATE_DOCTOR_ID        STRING," +
                "ORDERS_CREATE_TIME             STRING," +
                "ORDERS_CREATE_DEPT_NAME        STRING," +
                "ORDERS_CREATE_DOCTOR_NAME      STRING," +
                "ORDER_CODE                     STRING," +
                "ORDER_CONTENT                  STRING," +
                "ORDERS_COMMENTS                STRING," +
                "ORDER_TYPE                     STRING," +
                "ORDERS_START_TIME              STRING," +
                "ORDERS_END_TIME                STRING," +
                "ASSIST_EXAM_ITEM               STRING," +
                "ASSIST_EXAM_RESULT             STRING," +
                "XDS_TYPE                       STRING," +
                "XDS_NAME                       STRING," +
                "XDS_VERSION                    STRING," +
                "VISIT_ID                       STRING," +
                "table_name                     STRING," +
                "primary key (RECORD_ID) not enforced" +
                ")" +
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_EO_MR));

        tableEnv.executeSql("create table target2 (" +
                "RECIPEL_ID                      STRING," +
                "HOSPITAL_NAME                   STRING," +
                "HOSPITAL_CODE                   STRING," +
                "AGE_UNIT                        STRING," +
                "AGE                             STRING," +
                "SEX_NAME                        STRING," +
                "SEX_CODE                        STRING," +
                "ID_CARD_NUMBER                  STRING," +
                "PATIENT_VISIT_CATEGORY_CODE     STRING," +
                "PATIENT_VISIT_CATEGORY          STRING," +
                "OUTPATIENT_VISIT_FLOW_ID        STRING," +
                "OUTPATIENT_NO                   STRING," +
                "PATIENT_ID                      STRING," +
                "PATIENT_NAME                    STRING," +
                "DOMAIN_ID                       STRING," +
                "VISIT_TIMES                     STRING," +
                "OPEN_DEPT_CODE                  STRING," +
                "OPEN_DEPT                       STRING," +
                "AUTHOR_CODE                     STRING," +
                "AUTHOR_NAME                     STRING," +
                "CUSTODIAN_CODE                  STRING," +
                "CUSTODIAN_NAME                  STRING," +
                "AUDIT_SIGNDATE                  STRING," +
                "AUDIT_CODE                      STRING," +
                "AUDIT_NAME                      STRING," +
                "DEPLOY_SIGNDATE                 STRING," +
                "DEPLOY_CODE                     STRING," +
                "DEPLOY_NAME                     STRING," +
                "CHECK_SIGNDATE                  STRING," +
                "CHECK_CODE                      STRING," +
                "CHECK_NAME                      STRING," +
                "DISPENSER_SIGNDATE              STRING," +
                "DISPENSER_CODE                  STRING," +
                "DISPENSER_NAME                  STRING," +
                "RECIPEL_DAYS                    STRING," +
                "PDRUG_GROUP                     STRING," +
                "PDRUGS_FEE                      STRING," +
                "RECIPEL_REMARK                  STRING," +
                "VISIT_DATE                      STRING," +
                "RECIPE_FLAG                     STRING," +
                "DIAG_CODE                       STRING," +
                "DIAG_NAME                       STRING," +
                "XDS_TYPE                         STRING," +
                "XDS_NAME                         STRING," +
                "XDS_VERSION                      STRING," +
                "VISIT_ID                        STRING," +
                "table_name                     STRING," +
                "primary key (RECIPEL_ID) not enforced" +
                ")" +
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_PRESC));

        tableEnv.executeSql("create table target3(" +
                "RECIPEL_ID                       STRING," +
                "HOSPITAL_NAME                    STRING," +
                "HOSPITAL_CODE                    STRING," +
                "AGE_UNIT                         STRING," +
                "AGE                              STRING," +
                "SEX_NAME                         STRING," +
                "SEX_CODE                         STRING," +
                "ID_CARD_NUMBER                   STRING," +
                "PATIENT_VISIT_CATEGORY_CODE      STRING," +
                "PATIENT_VISIT_CATEGORY           STRING," +
                "OUTPATIENT_VISIT_FLOW_ID         STRING," +
                "OUTPATIENT_NO                    STRING," +
                "PATIENT_ID                       STRING," +
                "PATIENT_NAME                     STRING," +
                "DOMAIN_ID                        STRING," +
                "VISIT_TIMES                      STRING," +
                "OPEN_DEPT_CODE                   STRING," +
                "OPEN_DEPT                        STRING," +
                "AUTHOR_CODE                      STRING," +
                "AUTHOR_NAME                      STRING," +
                "CUSTODIAN_CODE                   STRING," +
                "CUSTODIAN_NAME                   STRING," +
                "AUDIT_SIGNDATE                   STRING," +
                "AUDIT_CODE                       STRING," +
                "AUDIT_NAME                       STRING," +
                "DEPLOY_SIGNDATE                  STRING," +
                "DEPLOY_CODE                      STRING," +
                "DEPLOY_NAME                      STRING," +
                "CHECK_SIGNDATE                   STRING," +
                "CHECK_CODE                       STRING," +
                "CHECK_NAME                       STRING," +
                "DISPENSER_SIGNDATE               STRING," +
                "DISPENSER_CODE                   STRING," +
                "DISPENSER_NAME                   STRING," +
                "RECIPEL_DAYS                     STRING," +
                "PDRUG_GROUP                      STRING," +
                "TCM_PIECES                       STRING," +
                "TCM_NUM_TABLETS                  STRING," +
                "TCM_METHOD                       STRING," +
                "CHINA_USED_METHOD                STRING," +
                "PCATEGORY_CODE                   STRING," +
                "PCATEGORY_NAME                   STRING," +
                "THERAPEUTIC_PRINCIPLE            STRING," +
                "PDRUGS_FEE                       STRING," +
                "RECIPEL_REMARK                   STRING," +
                "VISIT_DATE                       STRING," +
                "RECIPE_FLAG                      STRING," +
                "DIAG_CODE                        STRING," +
                "DIAG_NAME                        STRING," +
                "CHINA_DISEASE_CODE               STRING," +
                "CHINA_DISEASE_NAME               STRING," +
                "CHINA_SYNDROME_CODE              STRING," +
                "CHINA_SYNDROME_NAME              STRING," +
                "XDS_TYPE                         STRING," +
                "XDS_NAME                         STRING," +
                "XDS_VERSION                      STRING," +
                "VISIT_ID                         STRING," +
                "table_name                     STRING," +
                "primary key (RECIPEL_ID) not enforced" +
                ")" +
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_PRESC));

        tableEnv.executeSql("create table target4 (" +
                "XDS_TYPE                       STRING," +
                "XDS_NAME                       STRING," +
                "XDS_VERSION                    STRING," +
                "PATIENT_DOMAIN                 STRING," +
                "VISIT_DOMAIN                   STRING," +
                "CUSTODIAN_CODE                 STRING," +
                "CUSTODIAN_NAME                 STRING," +
                "HOSPITAL_CODE                  STRING," +
                "HOSPITAL_NAME                  STRING," +
                "VISIT_ID                       STRING," +
                "INPATIENT_VISIT_FLOW_ID        STRING," +
                "OUTPATIENT_VISIT_FLOW_ID       STRING," +
                "VISIT_TIMES                    STRING," +
                "PATIENT_ID                     STRING," +
                "RECORD_ID                      STRING," +
                "DW_DATE                        STRING," +
                "DOMAIN_ID                      STRING," +
                "OUTPATIENT_NO                  STRING," +
                "INPATIENT_NO                   STRING," +
                "PATIENT_VISIT_CATEGORY_CODE    STRING," +
                "PATIENT_VISIT_CATEGORY         STRING," +
                "EXAM_REPORT_NO                 STRING," +
                "APPLY_SHEET_NO                 STRING," +
                "SPECIMEN_NUMBER                STRING," +
                "TELECOM                        STRING," +
                "ID_CARD_NUMBER                 STRING," +
                "PATIENT_NAME                   STRING," +
                "SEX_NAME                       STRING," +
                "SEX_CODE                       STRING," +
                "AGE                            STRING," +
                "AGE_UNIT                       STRING," +
                "EXAM_REPORT_TIME               STRING," +
                "AUTHOR_CODE                    STRING," +
                "AUTHOR_NAME                    STRING," +
                "CUSTODIAN_ADDR                 STRING," +
                "AUDIT_SIGNDATE                 STRING," +
                "EXAM_AUDIT_DOCTOR_ID           STRING," +
                "EXAM_AUDIT_DOCTOR_NAME         STRING," +
                "EXAM_AUDIT_DATE                STRING," +
                "EXAM_TECH_STAFF_ID             STRING," +
                "EXAM_TECH_STAFF_NAME           STRING," +
                "TECHNICIAN_SIGNDATE            STRING," +
                "EXAM_DOCTOR_ID                 STRING," +
                "EXAM_DOCTOR_NAME               STRING," +
                "APPLY_DATE                     STRING," +
                "APPLY_DEPT_ID                  STRING," +
                "APPLY_DEPT                     STRING," +
                "APPLY_AGENCYCODE               STRING," +
                "APPLY_ORGAN                    STRING," +
                "VISIT_DATE                     STRING," +
                "BED_NO                         STRING," +
                "ROOM_NO                        STRING," +
                "ROOM_NAME                      STRING," +
                "WARDS_CODE                     STRING," +
                "BED_NAME                       STRING," +
                "DEPT_ID                        STRING," +
                "DEPT_NAME                      STRING," +
                "WARDS_NAME                     STRING," +
                "CHIEF_COMPLAINT                STRING," +
                "SYMPTOM_START_TIME             STRING," +
                "SYMPTOM_END_TIME               STRING," +
                "SYMPTOM_DESC                   STRING," +
                "EXAM_SPECIAL_FLAG              STRING," +
                "CHECK_METHOD                   STRING," +
                "INSPECT_TYPE                   STRING," +
                "TREAT_PROCESS_DESC             STRING," +
                "EXAM_REPORT_RESULT_FINDINGS    STRING," +
                "EXAM_REPORT_RESULT_HINT        STRING," +
                "EXAM_REPORT_DEPT_NAME          STRING," +
                "AGENCY_NAME                    STRING," +
                "REPORT_REMARK                  STRING," +
                "INSPECT_TYPE_CODE              STRING," +
                "EXAM_DATE                      STRING," +
                "ITEM_CODE                      STRING," +
                "ITEM_NAME                      STRING," +
                "SPECIMEN_SAMPLE_TIME           STRING," +
                "SPECIMEN_RECV_TIME             STRING," +
                "SPECIMEN_CLASS                 STRING," +
                "SPECIMEN_STATUS                STRING," +
                "SPECIMEN_FIXATIVE_NAME         STRING," +
                "RESULT_NAME                    STRING," +
                "RESULT_CODE                    STRING," +
                "EXAM_RESULT_VALUE              STRING," +
                "EXAM_RESULT_UNIT               STRING," +
                "DIAG_DATE                      STRING," +
                "DIAG_CODE                      STRING," +
                "DIAG_NAME                      STRING," +
                "DIAG_TYPE                      STRING," +
                "DIAG_INSTITUTION               STRING," +
                "primary key (RECORD_ID) not enforced  " +
                ")" +
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_EXAM_REPORT));

        // 导入数据
        StreamStatementSet statementSet = tableEnv.createStatementSet();
        statementSet.addInsertSql("insert into target1 select * from join_result1")
                .addInsertSql("insert into target2 select * from join_result2")
                .addInsertSql("insert into target3 select * from join_result3")
                .addInsertSql("insert into target4 select * from join_result4");
        statementSet.execute();
    }

    private void joinTables(StreamTableEnvironment tableEnv) {
        // 1 设置数据的TTL（暂时30min）
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60));
        // 2 join操作
        Table joinResult1 = tableEnv.sqlQuery("select " +
                "s.unique_id as RECORD_ID , " +
                "hospital_name as HOSPITAL_NAME , " +
                "hospital_code as HOSPITAL_CODE , " +
                "age_unit as AGE_UNIT , " +
                "age as AGE , " +
                "sex as SEX_NAME , " +
                "sex_code as SEX_CODE , " +
                "identify_no as ID_CARD_NUMBER , " +
                "patient_typecode as PATIENT_VISIT_CATEGORY_CODE , " +
                "patient_type as PATIENT_VISIT_CATEGORY , " +
                "visit_id as OUTPATIENT_VISIT_FLOW_ID , " +
                "out_patient_id as OUTPATIENT_NO , " +
                "patient_id as PATIENT_ID , " +
                "patient_name as PATIENT_NAME , " +
                "s.domain_id as DOMAIN_ID , " +
                "telecom as TELECOM , " +
                "visit_date as VISIT_DATE , " +
                "visit_times as VISIT_TIMES , " +
                "s.appy_id as APPLY_SHEET_NO , " +
                "birth as BIRTH_DATE , " +
                "dept_code as DEPT_ID , " +
                "dept_name as DEPT_NAME , " +
                "author_code as AUTHOR_CODE , " +
                "author_name as AUTHOR_NAME , " +
                "custodian_code as CUSTODIAN_CODE , " +
                "custodian_name as CUSTODIAN_NAME , " +
                "response_time as RESPONSE_TIME , " +
                "response_code as RESPONSE_CODE , " +
                "response_name as RESPONSE_NAME , " +
                "cast(case when allergy_history_mark='1' then allergy_history_mark='true' else allergy_history_mark='false' end as string) as ALLERGIC_HISTORY_FLAG , " +
                "allergy_history as ALLERGIC_HISTORY , " +
                "chief_complaint as CHIEF_COMPLAINT , " +
                "current_history as PRESENT_ILLNESS_HISTORY , " +
                "past_history as PAST_ILLNESS_HISTORY , " +
                "physical_examination as PHYSICAL_EXAM , " +
                "pd_code as FIRST_VISIT_FLAG , " +
                "pd_name as PD_NAME , " +
                "tcm_four_diagnosis as CHINA_MED_OBSERVE_RESULT , " +
                "discriminate_basis as DISCRIMINATE_BASIS , " +
                "rules_code as RULES_CODE , " +
                "rescue_start_date as RESCUE_START_DATE , " +
                "rescue_end_date as RESCUE_END_DATE , " +
                "rescuers_list as RESCUERS_LIST , " +
                "pat_category_code as PAT_CATEGORY_CODE , " +
                "pat_category as PAT_CATEGORY , " +
                "income_observatory_date as INCOME_OBSERVATORY_DATE , " +
                "emergency_course as EMERGENCY_COURSE , " +
                "needing_attention as NEEDING_ATTENTION , " +
                "destination_code as DESTINATION_CODE , " +
                "emergency_record as EMERGENCY_RECORD , " +
                "case when diag_type='西医诊断' then diag_code else '' end as DIAG_CODE , " +
                "case when diag_type='西医诊断' then diag_name else '' end as DIAG_NAME , " +
                "case when diag_type='中医病名' then diag_code else '' end as CHINA_DISEASE_CODE , " +
                "case when diag_type='中医病名' then diag_name else '' end as CHINA_DISEASE_NAME , " +
                "case when diag_type='中医证候' then diag_code else '' end as CHINA_SYNDROME_CODE , " +
                "case when diag_type='中医证候' then diag_name else '' end as CHINA_SYNDROME_NAME , " +
                "audit_code as ORDERS_AUDIT_STAFF_ID , " +
                "audit_date as ORDERS_AUDIT_TIME , " +
                "audit_order as ORDERS_AUDIT_STAFF_NAME , " +
                "cancel_code as ORDERS_CANCEL_DOCTOR_ID , " +
                "cancel_date as ORDERS_CANCEL_TIME , " +
                "cancel_name as ORDERS_CANCEL_DOCTOR_NAME , " +
                "executor_code as ORDERS_EXEC_STAFF_ID , " +
                "executor_name as ORDERS_EXEC_STAFF_NAME , " +
                "executor_date as ORDERS_EXEC_TIME , " +
                "executor_dept as ORDER_EXEC_DEPT , " +
                "executor_status as EXECUTOR_STATUS , " +
                "open_code as ORDERS_CREATE_DOCTOR_ID , " +
                "open_date as ORDERS_CREATE_TIME , " +
                "open_dept as ORDERS_CREATE_DEPT_NAME , " +
                "open_name as ORDERS_CREATE_DOCTOR_NAME , " +
                "order_code as ORDER_CODE , " +
                "order_content as ORDER_CONTENT , " +
                "order_remark as ORDERS_COMMENTS , " +
                "order_type as ORDER_TYPE , " +
                "start_date as ORDERS_START_TIME , " +
                "end_date as ORDERS_END_TIME , " +
                "item_name as ASSIST_EXAM_ITEM , " +
                "result_name as ASSIST_EXAM_RESULT," +
                "s.xds_type as XDS_TYPE," +
                "s.xds_name as XDS_NAME," +
                "s.xds_version as XDS_VERSION," +
                "visit_id  as VISIT_ID  ," +
                "'dwd_mhs_eo_mr' as table_name " +
                "from hdsd00_03_01 s    " +
                "left join tdy_list_diag d on s.unique_id = d.unique_id and s.xds_type = d.xds_type    " +
                "left join tdy_list_order o on s.unique_id = o.unique_id and s.xds_type = o.xds_type    " +
                "left join tdy_list_lexam l on s.unique_id = l.unique_id and s.xds_type = l.xds_type    ");
        tableEnv.createTemporaryView("join_result1",joinResult1);

        Table joinResult2 = tableEnv.sqlQuery("select " +
                "s.unique_id as RECIPEL_ID , " +
                "hospital_name as HOSPITAL_NAME , " +
                "hospital_code as HOSPITAL_CODE , " +
                "age_unit as AGE_UNIT , " +
                "age as AGE , " +
                "sex as SEX_NAME , " +
                "sex_code as SEX_CODE , " +
                "identify_no as ID_CARD_NUMBER , " +
                "patient_typecode as PATIENT_VISIT_CATEGORY_CODE , " +
                "patient_type as PATIENT_VISIT_CATEGORY , " +
                "visit_id as OUTPATIENT_VISIT_FLOW_ID , " +
                "out_patient_id as OUTPATIENT_NO , " +
                "patient_id as PATIENT_ID , " +
                "patient_name as PATIENT_NAME , " +
                "s.domain_id as DOMAIN_ID , " +
                "visit_times as VISIT_TIMES , " +
                "open_dept_code as OPEN_DEPT_CODE , " +
                "open_dept as OPEN_DEPT , " +
                "author_code as AUTHOR_CODE , " +
                "author_name as AUTHOR_NAME , " +
                "custodian_code as CUSTODIAN_CODE , " +
                "custodian_name as CUSTODIAN_NAME , " +
                "audit_signdate as AUDIT_SIGNDATE , " +
                "audit_code as AUDIT_CODE , " +
                "audit_name as AUDIT_NAME , " +
                "deploy_signdate as DEPLOY_SIGNDATE , " +
                "deploy_code as DEPLOY_CODE , " +
                "deploy_name as DEPLOY_NAME , " +
                "check_signdate as CHECK_SIGNDATE , " +
                "check_code as CHECK_CODE , " +
                "check_name as CHECK_NAME , " +
                "dispenser_signdate as DISPENSER_SIGNDATE , " +
                "dispenser_code as DISPENSER_CODE , " +
                "dispenser_name as DISPENSER_NAME , " +
                "recipel_days as RECIPEL_DAYS , " +
                "pdrug_group as PDRUG_GROUP , " +
                "pdrugs_fee as PDRUGS_FEE , " +
                "recipel_remark as RECIPEL_REMARK , " +
                "visit_date as VISIT_DATE , " +
                "'0' as RECIPE_FLAG , " +
                "diag_code as DIAG_CODE , " +
                "diag_name as DIAG_NAME," +
                "s.xds_type as XDS_TYPE," +
                "s.xds_name as XDS_NAME," +
                "s.xds_version as XDS_VERSION," +
                "visit_id as VISIT_ID  ," +
                "'dwd_mhs_presc' as table_name " +
                "from hdsd00_04_01 s    " +
                "left join tdy_list_diag d on s.unique_id = d.unique_id and s.xds_type = d.xds_type");
        tableEnv.createTemporaryView("join_result2",joinResult2);

        Table joinResult3 = tableEnv.sqlQuery("select " +
                "s.unique_id as RECIPEL_ID , " +
                "hospital_name as HOSPITAL_NAME , " +
                "hospital_code as HOSPITAL_CODE , " +
                "age_unit as AGE_UNIT , " +
                "age as AGE , " +
                "sex as SEX_NAME , " +
                "sex_code as SEX_CODE , " +
                "identify_no as ID_CARD_NUMBER , " +
                "patient_typecode as PATIENT_VISIT_CATEGORY_CODE , " +
                "patient_type as PATIENT_VISIT_CATEGORY , " +
                "visit_id as OUTPATIENT_VISIT_FLOW_ID , " +
                "out_patient_id as OUTPATIENT_NO , " +
                "patient_id as PATIENT_ID , " +
                "patient_name as PATIENT_NAME , " +
                "s.domain_id as DOMAIN_ID , " +
                "visit_times as VISIT_TIMES , " +
                "open_dept_code as OPEN_DEPT_CODE , " +
                "open_dept as OPEN_DEPT , " +
                "author_code as AUTHOR_CODE , " +
                "author_name as AUTHOR_NAME , " +
                "custodian_code as CUSTODIAN_CODE , " +
                "custodian_name as CUSTODIAN_NAME , " +
                "audit_signdate as AUDIT_SIGNDATE , " +
                "audit_code as AUDIT_CODE , " +
                "audit_name as AUDIT_NAME , " +
                "deploy_signdate as DEPLOY_SIGNDATE , " +
                "deploy_code as DEPLOY_CODE , " +
                "deploy_name as DEPLOY_NAME , " +
                "check_signdate as CHECK_SIGNDATE , " +
                "check_code as CHECK_CODE , " +
                "check_name as CHECK_NAME , " +
                "dispenser_signdate as DISPENSER_SIGNDATE , " +
                "dispenser_code as DISPENSER_CODE , " +
                "dispenser_name as DISPENSER_NAME , " +
                "recipel_days as RECIPEL_DAYS , " +
                "pdrug_group as PDRUG_GROUP , " +
                "tcm_pieces as TCM_PIECES , " +
                "tcm_num_tablets as TCM_NUM_TABLETS , " +
                "tcm_method as TCM_METHOD , " +
                "tcm_drug_use as CHINA_USED_METHOD , " +
                "pcategory_code as PCATEGORY_CODE , " +
                "pcategory_name as PCATEGORY_NAME , " +
                "principle_and_method as THERAPEUTIC_PRINCIPLE , " +
                "pdrugs_fee as PDRUGS_FEE , " +
                "recipel_remark as RECIPEL_REMARK , " +
                "visit_date as VISIT_DATE , " +
                "'1' as RECIPE_FLAG , " +
                "case when diag_type='西医诊断' then diag_code else '' end as DIAG_CODE , " +
                "case when diag_type='西医诊断' then diag_name else '' end as DIAG_NAME , " +
                "case when diag_type='中医病名' then diag_code else '' end as CHINA_DISEASE_CODE , " +
                "case when diag_type='中医病名' then diag_name else '' end as CHINA_DISEASE_NAME , " +
                "case when diag_type='中医证候' then diag_code else '' end as CHINA_SYNDROME_CODE , " +
                "case when diag_type='中医证候' then diag_name else '' end as CHINA_SYNDROME_NAME ," +
                "s.xds_type as XDS_TYPE," +
                "s.xds_name as XDS_NAME," +
                "s.xds_version as XDS_VERSION," +
                "visit_id as VISIT_ID," +
                "'dwd_mhs_presc' as table_name  " +
                "from hdsd00_04_02 s    " +
                "left join tdy_list_diag d on s.unique_id = d.unique_id and s.xds_type = d.xds_type    ");
        tableEnv.createTemporaryView("join_result3",joinResult3);

        Table joinResult4 = tableEnv.sqlQuery("" +
                "select                                                " +
                " s.xds_type                  as XDS_TYPE            , " +
                " s.xds_name                  as XDS_NAME            , " +
                " s.xds_version               as XDS_VERSION         , " +
                " s.patient_domain            as PATIENT_DOMAIN      , " +
                " s.visit_domain              as VISIT_DOMAIN        , " +
                " s.custodian_code            as CUSTODIAN_CODE      , " +
                " s.custodian_name            as CUSTODIAN_NAME      , " +
                " s.hospital_code             as HOSPITAL_CODE       , " +
                " s.hospital_name             as HOSPITAL_NAME       , " +
                " s.visit_id                  as VISIT_ID            , " +
                " case when s.patient_typecode  = '3' then visit_id else '' end as INPATIENT_VISIT_FLOW_ID        , " +
                " case when s.patient_typecode <> '3' then visit_id else '' end as OUTPATIENT_VISIT_FLOW_ID       , " +
                " s.visit_times               as VISIT_TIMES                       , " +
                " s.patient_id                as PATIENT_ID                        , " +
                " s.unique_id                 as RECORD_ID                         , " +
                " s.effective_time            as DW_DATE                           , " +
                " s.domain_id                 as DOMAIN_ID                         , " +
                " s.out_patient_id            as OUTPATIENT_NO                     , " +
                " s.inpatient                 as INPATIENT_NO                      , " +
                " s.patient_typecode          as PATIENT_VISIT_CATEGORY_CODE       , " +
                " s.patient_type              as PATIENT_VISIT_CATEGORY            , " +
                " s.report_id                 as EXAM_REPORT_NO                    , " +
                " s.appy_id                   as APPLY_SHEET_NO                    , " +
                " s.specimen_number           as SPECIMEN_NUMBER                   , " +
                " s.telecom                   as TELECOM                           , " +
                " s.identify_no               as ID_CARD_NUMBER                    , " +
                " s.patient_name              as PATIENT_NAME                      , " +
                " s.sex                       as SEX_NAME                          , " +
                " s.sex_code                  as SEX_CODE                          , " +
                " s.age                       as AGE                               , " +
                " s.age_unit                  as AGE_UNIT                          , " +
                " s.write_time                as EXAM_REPORT_TIME                  , " +
                " s.author_code               as AUTHOR_CODE                       , " +
                " s.author_name               as AUTHOR_NAME                       , " +
                " s.custodian_addr            as CUSTODIAN_ADDR                    , " +
                " s.audit_signdate            as AUDIT_SIGNDATE                    , " +
                " s.audit_code                as EXAM_AUDIT_DOCTOR_ID              , " +
                " s.audit_name                as EXAM_AUDIT_DOCTOR_NAME            , " +
                " s.check_signdate            as EXAM_AUDIT_DATE                   , " +
                " s.technician_code           as EXAM_TECH_STAFF_ID                , " +
                " s.technician_name           as EXAM_TECH_STAFF_NAME              , " +
                " s.technician_signdate       as TECHNICIAN_SIGNDATE               , " +
                " s.check_code                as EXAM_DOCTOR_ID                    , " +
                " s.check_name                as EXAM_DOCTOR_NAME                  , " +
                " s.apply_date                as APPLY_DATE                        , " +
                " s.apply_dept_code           as APPLY_DEPT_ID                     , " +
                " s.apply_dept                as APPLY_DEPT                        , " +
                " s.apply_agencycode          as APPLY_AGENCYCODE                  , " +
                " s.apply_organ               as APPLY_ORGAN                       , " +
                " s.visit_date                as VISIT_DATE                        , " +
                " s.bed_no                    as BED_NO                            , " +
                " s.ward_id                   as ROOM_NO                           , " +
                " s.ward_name                 as ROOM_NAME                         , " +
                " s.wards_id                  as WARDS_CODE                        , " +
                " s.bed_name                  as BED_NAME                          , " +
                " s.dept_code                 as DEPT_ID                           , " +
                " s.dept_name                 as DEPT_NAME                         , " +
                " s.wards_name                as WARDS_NAME                        , " +
                " s.chief_complaint           as CHIEF_COMPLAINT                   , " +
                " s.symptom_start_date        as SYMPTOM_START_TIME                , " +
                " s.symptom_stopping_date     as SYMPTOM_END_TIME                  , " +
                " s.symptom_description       as SYMPTOM_DESC                      , " +
                " s.special_exam_mark         as EXAM_SPECIAL_FLAG                 , " +
                " s.check_method              as CHECK_METHOD                      , " +
                " s.inspect_type              as INSPECT_TYPE                      , " +
                " s.treatment_process         as TREAT_PROCESS_DESC                , " +
                " s.objective_observation     as EXAM_REPORT_RESULT_FINDINGS       , " +
                " s.subjective_tip            as EXAM_REPORT_RESULT_HINT           , " +
                " s.report_dept               as EXAM_REPORT_DEPT_NAME             , " +
                " s.agency_name               as AGENCY_NAME                       , " +
                " s.report_remark             as REPORT_REMARK                     , " +
                " s.inspect_type_code         as INSPECT_TYPE_CODE                 , " +
                " e.check_date                as EXAM_DATE                         , " +
                " e.item_code                 as ITEM_CODE                         , " +
                " e.item_name                 as ITEM_NAME                         , " +
                " e.sample_date               as SPECIMEN_SAMPLE_TIME              , " +
                " e.receipt_date              as SPECIMEN_RECV_TIME                , " +
                " e.specimen_category         as SPECIMEN_CLASS                    , " +
                " e.specimen_status           as SPECIMEN_STATUS                   , " +
                " e.specimen_fixative         as SPECIMEN_FIXATIVE_NAME            , " +
                " e.result_name               as RESULT_NAME                       , " +
                " e.result_code               as RESULT_CODE                       , " +
                " e.quanty_result             as EXAM_RESULT_VALUE                 , " +
                " e.quanty_unit               as EXAM_RESULT_UNIT                  , " +
                " d.diag_date                 as DIAG_DATE                         , " +
                " d.diag_code                 as DIAG_CODE                         , " +
                " d.diag_name                 as DIAG_NAME                         , " +
                " d.diag_type                 as DIAG_TYPE                         , " +
                " d.diag_institution          as DIAG_INSTITUTION                    " +
                "from hdsd00_05_02 s                                                 " +
                "left join tdy_list_exam e on s.unique_id = e.unique_id and s.xds_type = e.xds_type " +
                "left join tdy_list_diag d on s.unique_id = d.unique_id and s.xds_type = d.xds_type " +
                "");
        tableEnv.createTemporaryView("join_result4",joinResult4);
    }

    private void createSourceTables(StreamTableEnvironment tableEnv) {
        // hdsd00_03_01
        tableEnv.executeSql("create table hdsd00_03_01 (" +
                "pk                      string," +
                "upload_time             string," +
                "status                  string," +
                "empi                    string," +
                "encounter_id            string," +
                "telecom                 string," +
                "patient_id              string," +
                "patient_domain          string," +
                "visit_date              string," +
                "visit_domain            string," +
                "visit_id                string," +
                "visit_times             string," +
                "unique_id               string," +
                "xds_type                string," +
                "xds_name                string," +
                "effective_time          string," +
                "xds_version             string," +
                "out_patient_id          string," +
                "domain_id               string," +
                "appy_id                 string," +
                "patient_type            string," +
                "identify_no             string," +
                "patient_typecode        string," +
                "patient_name            string," +
                "sex_code                string," +
                "sex                     string," +
                "birth                   string," +
                "age                     string," +
                "age_unit                string," +
                "dept_code               string," +
                "dept_name               string," +
                "hospital_code           string," +
                "hospital_name           string," +
                "write_time              string," +
                "author_code             string," +
                "author_name             string," +
                "custodian_code          string," +
                "custodian_name          string," +
                "response_time           string," +
                "response_code           string," +
                "response_name           string," +
                "allergy_history_mark    string," +
                "allergy_history         string," +
                "chief_complaint         string," +
                "current_history         string," +
                "past_history            string," +
                "physical_examination    string," +
                "pd_code                 string," +
                "pd_name                 string," +
                "tcm_four_diagnosis      string," +
                "discriminate_basis      string," +
                "rules_code              string," +
                "rescue_start_date       string," +
                "rescue_end_date         string," +
                "rescuers_list           string," +
                "pat_category_code       string," +
                "pat_category            string," +
                "income_observatory_date string," +
                "emergency_course        string," +
                "needing_attention       string," +
                "destination_code        string," +
                "emergency_record        string," +
                "check_item              string," +
                "item_result             string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_03_01","App_01_04"));
        // hdsd00_04_01
        tableEnv.executeSql("create table hdsd00_04_01 (" +
                "pk                        string," +
                "upload_time               string," +
                "status                    string," +
                "empi                      string," +
                "encounter_id              string," +
                "patient_id                string," +
                "patient_domain            string," +
                "visit_domain              string," +
                "visit_id                  string," +
                "visit_times               string," +
                "unique_id                 string," +
                "xds_type                  string," +
                "xds_name                  string," +
                "effective_time            string," +
                "xds_version               string," +
                "out_patient_id            string," +
                "domain_id                 string," +
                "recipel_id                string," +
                "patient_type              string," +
                "identify_no               string," +
                "patient_typecode          string," +
                "patient_name              string," +
                "sex_code                  string," +
                "sex                       string," +
                "age                       string," +
                "age_unit                  string," +
                "open_dept_code            string," +
                "open_dept                 string," +
                "hospital_code             string," +
                "hospital_name             string," +
                "write_time                string," +
                "author_code               string," +
                "author_name               string," +
                "custodian_code            string," +
                "custodian_name            string," +
                "audit_signdate            string," +
                "audit_code                string," +
                "audit_name                string," +
                "check_signdate            string," +
                "deploy_code               string," +
                "deploy_name               string," +
                "deploy_signdate           string," +
                "check_code                string," +
                "check_name                string," +
                "dispenser_signdate        string," +
                "dispenser_code            string," +
                "dispenser_name            string," +
                "pdrugs_fee                string," +
                "recipel_days              string," +
                "pdrug_group               string," +
                "recipel_remark            string," +
                "visit_date                string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_04_01","App_01_04"));
        // hdsd00_04_02
        tableEnv.executeSql("create table hdsd00_04_02 (" +
                "pk                        string," +
                "upload_time               string," +
                "status                    string," +
                "empi                      string," +
                "encounter_id              string," +
                "patient_id                string," +
                "patient_domain            string," +
                "visit_domain              string," +
                "visit_id                  string," +
                "visit_times               string," +
                "unique_id                 string," +
                "xds_type                  string," +
                "xds_name                  string," +
                "effective_time            string," +
                "xds_version               string," +
                "out_patient_id            string," +
                "domain_id                 string," +
                "recipel_id                string," +
                "patient_type              string," +
                "identify_no               string," +
                "patient_typecode          string," +
                "patient_name              string," +
                "sex                       string," +
                "sex_code                  string," +
                "age                       string," +
                "age_unit                  string," +
                "open_dept_code            string," +
                "open_dept                 string," +
                "hospital_code             string," +
                "hospital_name             string," +
                "write_time                string," +
                "author_code               string," +
                "author_name               string," +
                "custodian_code            string," +
                "custodian_name            string," +
                "audit_signdate            string," +
                "audit_code                string," +
                "audit_name                string," +
                "check_signdate            string," +
                "deploy_code               string," +
                "deploy_name               string," +
                "deploy_signdate           string," +
                "check_code                string," +
                "check_name                string," +
                "dispenser_signdate        string," +
                "dispenser_code            string," +
                "dispenser_name            string," +
                "recipel_days              string," +
                "pdrug_group               string," +
                "tcm_pieces                string," +
                "tcm_num_tablets           string," +
                "tcm_method                string," +
                "tcm_drug_use              string," +
                "pcategory_code            string," +
                "pcategory_name            string," +
                "pdrugs_fee                string," +
                "recipel_remark            string," +
                "principle_and_method      string," +
                "visit_date                string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_04_02","App_01_04"));
        // hdsd00_05_02
        tableEnv.executeSql("create table hdsd00_05_02 (" +
                "pk                         string," +
                "upload_time                string," +
                "status                     string," +
                "empi                       string," +
                "encounter_id               string," +
                "visit_domain               string," +
                "visit_id                   string," +
                "visit_times                string," +
                "patient_id                 string," +
                "patient_domain             string," +
                "unique_id                  string," +
                "xds_type                   string," +
                "xds_name                   string," +
                "effective_time             string," +
                "xds_version                string," +
                "domain_id                  string," +
                "out_patient_id             string," +
                "inpatient                  string," +
                "patient_type               string," +
                "report_id                  string," +
                "patient_typecode           string," +
                "appy_id                    string," +
                "specimen_number            string," +
                "telecom                    string," +
                "identify_no                string," +
                "patient_name               string," +
                "sex                        string," +
                "sex_code                   string," +
                "age                        string," +
                "age_unit                   string," +
                "write_time                 string," +
                "author_code                string," +
                "author_name                string," +
                "custodian_name             string," +
                "custodian_code             string," +
                "custodian_addr             string," +
                "audit_signdate             string," +
                "audit_code                 string," +
                "audit_name                 string," +
                "check_signdate             string," +
                "technician_code            string," +
                "technician_name            string," +
                "technician_signdate        string," +
                "check_code                 string," +
                "check_name                 string," +
                "apply_date                 string," +
                "apply_dept_code            string," +
                "apply_dept                 string," +
                "apply_agencycode           string," +
                "apply_organ                string," +
                "visit_date                 string," +
                "bed_no                     string," +
                "ward_id                    string," +
                "ward_name                  string," +
                "wards_id                   string," +
                "bed_name                   string," +
                "dept_code                  string," +
                "dept_name                  string," +
                "wards_name                 string," +
                "hospital_code              string," +
                "hospital_name              string," +
                "chief_complaint            string," +
                "symptom_start_date         string," +
                "symptom_stopping_date      string," +
                "symptom_description        string," +
                "special_exam_mark          string," +
                "check_method               string," +
                "inspect_type               string," +
                "treatment_process          string," +
                "objective_observation      string," +
                "subjective_tip             string," +
                "report_dept                string," +
                "agency_name                string," +
                "report_remark              string," +
                "inspect_type_code          string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_05_02","App_01_04"));

        // tdy_list_diag
        super.readTdyListDiag(tableEnv,"App_01_04");
        // tdy_list_order
        super.readTdyListOrder(tableEnv,"App_01_04");
        // tdy_list_lexam
        super.readTdyListLexam(tableEnv,"App_01_04");
        // tdy_list_exam
        super.readTdyListExam(tableEnv,"App_01_04");
    }
}
